{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Kinesis Data Analytics for SQL Applications\n",
    "\n",
    "To get started with Kinesis Data Analytics, you create a Kinesis data analytics application that continuously reads and processes streaming data."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"img/use_case_1_analytics.png\" width=\"80%\" align=\"left\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"img/use_case_2_anomaly.png\" width=\"83%\" align=\"left\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"img/use_case_3_count.png\" width=\"80%\" align=\"left\">"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import boto3\n",
    "import sagemaker\n",
    "import pandas as pd\n",
    "import json\n",
    "\n",
    "sess = sagemaker.Session()\n",
    "bucket = sess.default_bucket()\n",
    "role = sagemaker.get_execution_role()\n",
    "region = boto3.Session().region_name\n",
    "\n",
    "sts = boto3.Session().client(service_name=\"sts\", region_name=region)\n",
    "account_id = sts.get_caller_identity()[\"Account\"]\n",
    "\n",
    "sm = boto3.Session().client(service_name=\"sagemaker\", region_name=region)\n",
    "firehose = boto3.Session().client(service_name=\"firehose\", region_name=region)\n",
    "kinesis_analytics = boto3.Session().client(service_name=\"kinesisanalytics\", region_name=region)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r firehose_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    firehose_arn\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(firehose_arn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r iam_role_kinesis_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    iam_role_kinesis_arn\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(iam_role_kinesis_arn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r stream_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    stream_arn\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(stream_arn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r lambda_fn_arn_cloudwatch"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    lambda_fn_arn_cloudwatch\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(lambda_fn_arn_cloudwatch)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r lambda_fn_arn_sns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    lambda_fn_arn_sns\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(lambda_fn_arn_sns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r iam_role_lambda_arn"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    iam_role_lambda_arn\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(iam_role_lambda_arn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store -r lambda_fn_arn_invoke_ep"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    lambda_fn_arn_invoke_ep\n",
    "except NameError:\n",
    "    print(\"+++++++++++++++++++++++++++++++\")\n",
    "    print(\"[ERROR] Please run all previous notebooks in this section before you continue.\")\n",
    "    print(\"+++++++++++++++++++++++++++++++\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(lambda_fn_arn_invoke_ep)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create a Kinesis Data Analytics for SQL Application\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define the Kinesis Analytics Application Name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "kinesis_data_analytics_app_name = \"dsoaws-kinesis-data-analytics-sql-app\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "in_app_stream_name = \"SOURCE_SQL_STREAM_001\"  # Default\n",
    "print(in_app_stream_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Application"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "window_seconds = 5"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sql_code = \"\"\" \\\n",
    "        CREATE OR REPLACE STREAM \"AVG_STAR_RATING_SQL_STREAM\" ( \\\n",
    "            avg_star_rating DOUBLE); \\\n",
    "        CREATE OR REPLACE PUMP \"AVG_STAR_RATING_SQL_STREAM_PUMP\" AS \\\n",
    "            INSERT INTO \"AVG_STAR_RATING_SQL_STREAM\" \\\n",
    "                SELECT STREAM AVG(CAST(\"star_rating\" AS DOUBLE)) AS avg_star_rating \\\n",
    "                FROM \"{}\" \\\n",
    "                GROUP BY \\\n",
    "                STEP(\"{}\".ROWTIME BY INTERVAL '{}' SECOND); \\\n",
    "         \\\n",
    "        CREATE OR REPLACE STREAM \"ANOMALY_SCORE_SQL_STREAM\" (anomaly_score DOUBLE); \\\n",
    "        CREATE OR REPLACE PUMP \"ANOMALY_SCORE_STREAM_PUMP\" AS \\\n",
    "            INSERT INTO \"ANOMALY_SCORE_SQL_STREAM\" \\\n",
    "            SELECT STREAM anomaly_score \\\n",
    "            FROM TABLE(RANDOM_CUT_FOREST( \\\n",
    "                CURSOR(SELECT STREAM \"star_rating\" \\\n",
    "                    FROM \"{}\" \\\n",
    "            ) \\\n",
    "          ) \\\n",
    "        ); \\\n",
    "         \\\n",
    "        CREATE OR REPLACE STREAM \"APPROXIMATE_COUNT_SQL_STREAM\" (number_of_distinct_items BIGINT); \\\n",
    "        CREATE OR REPLACE PUMP \"APPROXIMATE_COUNT_STREAM_PUMP\" AS \\\n",
    "            INSERT INTO \"APPROXIMATE_COUNT_SQL_STREAM\" \\\n",
    "            SELECT STREAM number_of_distinct_items \\\n",
    "            FROM TABLE(COUNT_DISTINCT_ITEMS_TUMBLING( \\\n",
    "                CURSOR(SELECT STREAM \"review_id\" FROM \"{}\"), \\\n",
    "                'review_id', \\\n",
    "                {} \\\n",
    "              ) \\\n",
    "        ); \\\n",
    "    \"\"\".format(\n",
    "    in_app_stream_name, in_app_stream_name, window_seconds, in_app_stream_name, in_app_stream_name, window_seconds\n",
    ")\n",
    "\n",
    "print(sql_code)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from botocore.exceptions import ClientError\n",
    "\n",
    "try:\n",
    "    response = kinesis_analytics.create_application(\n",
    "        ApplicationName=kinesis_data_analytics_app_name,\n",
    "        Inputs=[\n",
    "            {\n",
    "                \"NamePrefix\": \"SOURCE_SQL_STREAM\",\n",
    "                \"KinesisFirehoseInput\": {\n",
    "                    \"ResourceARN\": \"{}\".format(firehose_arn),\n",
    "                    \"RoleARN\": \"{}\".format(iam_role_kinesis_arn),\n",
    "                },\n",
    "                \"InputProcessingConfiguration\": {\n",
    "                    \"InputLambdaProcessor\": {\n",
    "                        \"ResourceARN\": \"{}\".format(lambda_fn_arn_invoke_ep),\n",
    "                        \"RoleARN\": \"{}\".format(iam_role_lambda_arn),\n",
    "                    }\n",
    "                },\n",
    "                \"InputSchema\": {\n",
    "                    \"RecordFormat\": {\n",
    "                        \"RecordFormatType\": \"CSV\",\n",
    "                        \"MappingParameters\": {\n",
    "                            \"CSVMappingParameters\": {\"RecordRowDelimiter\": \"\\n\", \"RecordColumnDelimiter\": \"\\t\"}\n",
    "                        },\n",
    "                    },\n",
    "                    \"RecordColumns\": [\n",
    "                        {\"Name\": \"review_id\", \"Mapping\": \"review_id\", \"SqlType\": \"VARCHAR(14)\"},\n",
    "                        {\"Name\": \"star_rating\", \"Mapping\": \"star_rating\", \"SqlType\": \"INTEGER\"},\n",
    "                        {\"Name\": \"product_category\", \"Mapping\": \"product_category\", \"SqlType\": \"VARCHAR(24)\"},\n",
    "                        {\"Name\": \"review_body\", \"Mapping\": \"review_body\", \"SqlType\": \"VARCHAR(65535)\"},\n",
    "                    ],\n",
    "                },\n",
    "            },\n",
    "        ],\n",
    "        Outputs=[\n",
    "            {\n",
    "                \"Name\": \"AVG_STAR_RATING_SQL_STREAM\",\n",
    "                \"LambdaOutput\": {\n",
    "                    \"ResourceARN\": \"{}\".format(lambda_fn_arn_cloudwatch),\n",
    "                    \"RoleARN\": \"{}\".format(iam_role_lambda_arn),\n",
    "                },\n",
    "                \"DestinationSchema\": {\"RecordFormatType\": \"CSV\"},\n",
    "            },\n",
    "            {\n",
    "                \"Name\": \"ANOMALY_SCORE_SQL_STREAM\",\n",
    "                \"LambdaOutput\": {\n",
    "                    \"ResourceARN\": \"{}\".format(lambda_fn_arn_sns),\n",
    "                    \"RoleARN\": \"{}\".format(iam_role_kinesis_arn),\n",
    "                },\n",
    "                \"DestinationSchema\": {\"RecordFormatType\": \"CSV\"},\n",
    "            },\n",
    "            {\n",
    "                \"Name\": \"APPROXIMATE_COUNT_SQL_STREAM\",\n",
    "                \"KinesisStreamsOutput\": {\n",
    "                    \"ResourceARN\": \"{}\".format(stream_arn),\n",
    "                    \"RoleARN\": \"{}\".format(iam_role_kinesis_arn),\n",
    "                },\n",
    "                \"DestinationSchema\": {\"RecordFormatType\": \"CSV\"},\n",
    "            },\n",
    "        ],\n",
    "        ApplicationCode=sql_code,\n",
    "    )\n",
    "    print(\"SQL application {} successfully created.\".format(kinesis_data_analytics_app_name))\n",
    "    print(json.dumps(response, indent=4, sort_keys=True, default=str))\n",
    "except ClientError as e:\n",
    "    if e.response[\"Error\"][\"Code\"] == \"ResourceInUseException\":\n",
    "        print(\"SQL App {} already exists.\".format(kinesis_data_analytics_app_name))\n",
    "    else:\n",
    "        print(\"Unexpected error: %s\" % e)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "response = kinesis_analytics.describe_application(ApplicationName=kinesis_data_analytics_app_name)\n",
    "print(json.dumps(response, indent=4, sort_keys=True, default=str))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_id = response[\"ApplicationDetail\"][\"InputDescriptions\"][0][\"InputId\"]\n",
    "print(input_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Start the Kinesis Data Analytics App"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "try:\n",
    "    response = kinesis_analytics.start_application(\n",
    "        ApplicationName=kinesis_data_analytics_app_name,\n",
    "        InputConfigurations=[{\"Id\": input_id, \"InputStartingPositionConfiguration\": {\"InputStartingPosition\": \"NOW\"}}],\n",
    "    )\n",
    "    print(json.dumps(response, indent=4, sort_keys=True, default=str))\n",
    "except ClientError as e:\n",
    "    if e.response[\"Error\"][\"Code\"] == \"ResourceInUseException\":\n",
    "        print(\"Application {} is already starting.\".format(kinesis_data_analytics_app_name))\n",
    "    else:\n",
    "        print(\"Error: {}\".format(e))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%store kinesis_data_analytics_app_name"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Explore Kinesis Data Analytics App"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from IPython.core.display import display, HTML\n",
    "\n",
    "display(\n",
    "    HTML(\n",
    "        '<b>Review <a target=\"blank\" href=\"https://console.aws.amazon.com/kinesisanalytics/home?region={}#/wizard/hub?applicationName={}\"> Kinesis Data Analytics App</a></b>'.format(\n",
    "            region, kinesis_data_analytics_app_name\n",
    "        )\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "response = kinesis_analytics.describe_application(ApplicationName=kinesis_data_analytics_app_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "import time\n",
    "\n",
    "app_status = response[\"ApplicationDetail\"][\"ApplicationStatus\"]\n",
    "print(\"Application status {}\".format(app_status))\n",
    "\n",
    "while app_status != \"RUNNING\":\n",
    "    time.sleep(5)\n",
    "    response = kinesis_analytics.describe_application(ApplicationName=kinesis_data_analytics_app_name)\n",
    "    app_status = response[\"ApplicationDetail\"][\"ApplicationStatus\"]\n",
    "    print(\"Application status {}\".format(app_status))\n",
    "\n",
    "print(\"Application status {}\".format(app_status))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# _Please be patient.  ^^ This may take a few minutes. ^^_"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Store Variables for Next Notebooks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Release Resources"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%html\n",
    "\n",
    "<p><b>Shutting down your kernel for this notebook to release resources.</b></p>\n",
    "<button class=\"sm-command-button\" data-commandlinker-command=\"kernelmenu:shutdown\" style=\"display:none;\">Shutdown Kernel</button>\n",
    "        \n",
    "<script>\n",
    "try {\n",
    "    els = document.getElementsByClassName(\"sm-command-button\");\n",
    "    els[0].click();\n",
    "}\n",
    "catch(err) {\n",
    "    // NoOp\n",
    "}    \n",
    "</script>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%javascript\n",
    "\n",
    "try {\n",
    "    Jupyter.notebook.save_checkpoint();\n",
    "    Jupyter.notebook.session.delete();\n",
    "}\n",
    "catch(err) {\n",
    "    // NoOp\n",
    "}"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
